# -*- coding: utf-8 -*-
"""
人寿保险盈利能力指标计算
工单编号：大数据-八维保险数据挖掘-07-人寿保险盈利能力
功能：实现文档中所有保险盈利能力指标的计算，支持按七日、月、年统计
"""

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, avg, sum, when, datediff, months_between, year, month, date_add
from pyspark.sql.window import Window
import pyspark.sql.functions as F
from datetime import datetime, timedelta

# 初始化SparkSession
spark = SparkSession.builder \
    .appName("InsuranceProfitabilityAnalysis") \
    .enableHiveSupport() \
    .getOrCreate()


# 数据加载函数
def load_data(table_name):
    """加载指定表的数据"""
    return spark.table(f"insurance_db.{table_name}")


# 加载各表数据
org_df = load_data("organization")
policy_df = load_data("policy")
reinsurance_df = load_data("reinsurance_outgoing")
financial_df = load_data("financial_data")
balance_sheet_df = load_data("balance_sheet")
underwriting_df = load_data("underwriting_data")


# 工具函数：获取期初和期末数据
def get_period_data(df, date_col, value_col, org_id_col, start_date, end_date):
    """获取指定期间的期初和期末数据"""
    # 期初数据（期间开始时的数据）
    start_window = Window.partitionBy(org_id_col).orderBy(date_col)
    start_df = df.filter(col(date_col) <= start_date) \
        .withColumn("rn", F.row_number().over(start_window.orderBy(col(date_col).desc()))) \
        .filter(col("rn") == 1) \
        .select(col(org_id_col).alias("org_id"),
                col(value_col).alias(f"start_{value_col}"))

    # 期末数据（期间结束时的数据）
    end_window = Window.partitionBy(org_id_col).orderBy(date_col)
    end_df = df.filter(col(date_col) <= end_date) \
        .withColumn("rn", F.row_number().over(end_window.orderBy(col(date_col).desc()))) \
        .filter(col("rn") == 1) \
        .select(col(org_id_col).alias("org_id"),
                col(value_col).alias(f"end_{value_col}"))

    return start_df.join(end_df, on="org_id", how="inner")


# 1. 净资产收益率计算
def calculate_roe(start_date, end_date, period_type="month"):
    """
    计算净资产收益率
    净资产收益率 = 报告期净利润 ÷[(期初所有者权益 + 期末所有者权益)÷2]×100%
    """
    # 获取所有者权益期初和期末数据
    equity_df = get_period_data(balance_sheet_df, "data_date", "owner_equity", "org_id", start_date, end_date)

    # 获取报告期净利润
    profit_df = financial_df.filter((col("data_date") >= start_date) & (col("data_date") <= end_date)) \
        .groupBy("org_id") \
        .agg(sum("net_profit").alias("period_net_profit"))

    # 计算净资产收益率
    roe_df = equity_df.join(profit_df, on="org_id", how="inner") \
        .withColumn("avg_owner_equity", (col("start_owner_equity") + col("end_owner_equity")) / 2) \
        .withColumn("roe", when(col("avg_owner_equity") != 0,
                                (col("period_net_profit") / col("avg_owner_equity")) * 100)
                    .otherwise(0)) \
        .select("org_id", "roe",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return roe_df


# 2. 总资产收益率计算
def calculate_roa(start_date, end_date, period_type="month"):
    """
    计算总资产收益率
    总资产收益率 = 报告期净利润 ÷[(期初总资产 + 期末总资产)÷2]×100%
    """
    # 获取总资产期初和期末数据
    assets_df = get_period_data(balance_sheet_df, "data_date", "total_assets", "org_id", start_date, end_date)

    # 获取报告期净利润
    profit_df = financial_df.filter((col("data_date") >= start_date) & (col("data_date") <= end_date)) \
        .groupBy("org_id") \
        .agg(sum("net_profit").alias("period_net_profit"))

    # 计算总资产收益率
    roa_df = assets_df.join(profit_df, on="org_id", how="inner") \
        .withColumn("avg_total_assets", (col("start_total_assets") + col("end_total_assets")) / 2) \
        .withColumn("roa", when(col("avg_total_assets") != 0,
                                (col("period_net_profit") / col("avg_total_assets")) * 100)
                    .otherwise(0)) \
        .select("org_id", "roa",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return roa_df


# 3. 总资产周转率计算
def calculate_asset_turnover(start_date, end_date, period_type="month"):
    """
    计算总资产周转率
    总资产周转率 = 报告期营业收入 ÷[(期初总资产 + 期末总资产)÷2]×100%
    其中营业收入 = 已赚保费
    """
    # 获取总资产期初和期末数据
    assets_df = get_period_data(balance_sheet_df, "data_date", "total_assets", "org_id", start_date, end_date)

    # 获取报告期营业收入（已赚保费）
    income_df = policy_df.filter((col("data_date") >= start_date) & (col("data_date") <= end_date)) \
        .groupBy("org_id") \
        .agg(sum("earned_premium").alias("operating_income"))

    # 计算总资产周转率
    turnover_df = assets_df.join(income_df, on="org_id", how="inner") \
        .withColumn("avg_total_assets", (col("start_total_assets") + col("end_total_assets")) / 2) \
        .withColumn("asset_turnover", when(col("avg_total_assets") != 0,
                                           (col("operating_income") / col("avg_total_assets")) * 100)
                    .otherwise(0)) \
        .select("org_id", "asset_turnover",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return turnover_df


# 4. 应收保费率计算
def calculate_receivable_premium_rate(start_date, end_date, period_type="month"):
    """
    计算应收保费率
    应收保费率 = 账龄不长的应收保费余额 ÷ 滚动12个月原保费收入 ×100%
    """
    # 计算滚动12个月的起始日期
    roll_12_start = (datetime.strptime(start_date, "%Y-%m-%d") - timedelta(days=365)).strftime("%Y-%m-%d")

    # 获取账龄不长的应收保费余额（期末）
    short_age_df = balance_sheet_df.filter(col("data_date") == end_date) \
        .select("org_id", "short_age_receivable_premium")

    # 获取滚动12个月原保费收入
    roll_premium_df = policy_df.filter((col("data_date") >= roll_12_start) & (col("data_date") <= end_date)) \
        .groupBy("org_id") \
        .agg(sum("original_premium").alias("roll_12_original_premium"))

    # 计算应收保费率
    rate_df = short_age_df.join(roll_premium_df, on="org_id", how="inner") \
        .withColumn("receivable_premium_rate", when(col("roll_12_original_premium") != 0,
                                                    (col("short_age_receivable_premium") / col(
                                                        "roll_12_original_premium")) * 100)
                    .otherwise(0)) \
        .select("org_id", "receivable_premium_rate",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return rate_df


# 5. 应收保费周转率计算
def calculate_receivable_turnover(start_date, end_date, period_type="month"):
    """
    计算应收保费周转率
    应收保费周转率 = 报告期原保费收入 ÷[(期初应收保费余额 + 期末应收保费余额)÷2]×100%
    """
    # 获取应收保费期初和期末数据
    receivable_df = get_period_data(balance_sheet_df, "data_date", "receivable_premium", "org_id", start_date, end_date)

    # 获取报告期原保费收入
    premium_df = policy_df.filter((col("data_date") >= start_date) & (col("data_date") <= end_date)) \
        .groupBy("org_id") \
        .agg(sum("original_premium").alias("period_original_premium"))

    # 计算应收保费周转率
    turnover_df = receivable_df.join(premium_df, on="org_id", how="inner") \
        .withColumn("avg_receivable_premium", (col("start_receivable_premium") + col("end_receivable_premium")) / 2) \
        .withColumn("receivable_turnover", when(col("avg_receivable_premium") != 0,
                                                (col("period_original_premium") / col("avg_receivable_premium")) * 100)
                    .otherwise(0)) \
        .select("org_id", "receivable_turnover",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return turnover_df


# 6. 内含价值计算
def calculate_embedded_value(start_date, end_date, period_type="month"):
    """
    计算内含价值
    内含价值 = 所有者权益 + 有效业务价值
    （假设有效业务价值从专门的表中获取，这里简化处理）
    """
    # 获取期末所有者权益
    equity_df = balance_sheet_df.filter(col("data_date") == end_date) \
        .select("org_id", "owner_equity")

    # 假设有效业务价值从专门的表中获取，这里简化处理
    # 实际应用中需要根据业务定义计算有效业务价值
    ev_df = spark.table("insurance_db.effective_business_value") \
        .filter(col("data_date") == end_date) \
        .select("org_id", "effective_value")

    # 计算内含价值
    embedded_value_df = equity_df.join(ev_df, on="org_id", how="inner") \
        .withColumn("embedded_value", col("owner_equity") + col("effective_value")) \
        .select("org_id", "embedded_value",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return embedded_value_df


# 7. 总资产增长率计算
def calculate_assets_growth_rate(start_date, end_date, period_type="month"):
    """
    计算总资产增长率
    总资产增长率 =(期末总资产 - 期初总资产)÷ 期初总资产 ×100%
    """
    # 获取总资产期初和期末数据
    assets_df = get_period_data(balance_sheet_df, "data_date", "total_assets", "org_id", start_date, end_date)

    # 计算总资产增长率
    growth_df = assets_df \
        .withColumn("assets_growth_rate", when(col("start_total_assets") != 0,
                                               ((col("end_total_assets") - col("start_total_assets")) / col(
                                                   "start_total_assets")) * 100)
                    .otherwise(0)) \
        .select("org_id", "assets_growth_rate",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return growth_df


# 8. 净资产增长率计算
def calculate_equity_growth_rate(start_date, end_date, period_type="month"):
    """
    计算净资产增长率
    净资产增长率 =(期末所有者权益 - 期初所有者权益)÷ 期初所有者权益 ×100%
    """
    # 获取所有者权益期初和期末数据
    equity_df = get_period_data(balance_sheet_df, "data_date", "owner_equity", "org_id", start_date, end_date)

    # 计算净资产增长率
    growth_df = equity_df \
        .withColumn("equity_growth_rate", when(col("start_owner_equity") != 0,
                                               ((col("end_owner_equity") - col("start_owner_equity")) / col(
                                                   "start_owner_equity")) * 100)
                    .otherwise(0)) \
        .select("org_id", "equity_growth_rate",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return growth_df


# 9. 承保利润率计算
def calculate_underwriting_profit_rate(start_date, end_date, period_type="month"):
    """
    计算承保利润率
    承保利润率 = 承保利润 ÷ 已赚保费 ×100%
    其中承保利润 = 已赚保费 - 赔付支出 + 分保费用收入 - 分保费用支出 - 手续费及佣金支出 - 业务及管理费 - 税金及附加
    """
    # 汇总报告期承保数据
    underwriting_summary_df = underwriting_df.filter((col("data_date") >= start_date) & (col("data_date") <= end_date)) \
        .groupBy("org_id") \
        .agg(
        sum("earned_premium").alias("total_earned_premium"),
        sum("compensation_expense").alias("total_compensation"),
        sum("reinsurance_income").alias("total_reinsurance_income"),
        sum("reinsurance_expense").alias("total_reinsurance_expense"),
        sum("commission_expense").alias("total_commission"),
        sum("business_management_fee").alias("total_management_fee"),
        sum("taxes_surcharges").alias("total_taxes")
    )

    # 计算承保利润和承保利润率
    profit_rate_df = underwriting_summary_df \
        .withColumn("underwriting_profit",
                    col("total_earned_premium") - col("total_compensation") +
                    col("total_reinsurance_income") - col("total_reinsurance_expense") -
                    col("total_commission") - col("total_management_fee") - col("total_taxes")) \
        .withColumn("underwriting_profit_rate", when(col("total_earned_premium") != 0,
                                                     (col("underwriting_profit") / col("total_earned_premium")) * 100)
                    .otherwise(0)) \
        .select("org_id", "underwriting_profit_rate",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return profit_rate_df


# 10. 自留保费占净资产比计算
def calculate_retained_premium_ratio(start_date, end_date, period_type="month"):
    """
    计算自留保费占净资产比
    自留保费占净资产比 = 报告期自留保费 ÷[(期初所有者权益 + 期末所有者权益)÷2]×100%
    自留保费 = 总保费 - 分出保费
    """
    # 获取所有者权益期初和期末数据
    equity_df = get_period_data(balance_sheet_df, "data_date", "owner_equity", "org_id", start_date, end_date)

    # 计算总保费
    total_premium_df = policy_df.filter((col("data_date") >= start_date) & (col("data_date") <= end_date)) \
        .groupBy("org_id") \
        .agg(sum("total_premium").alias("total_premium"))

    # 计算分出保费
    ceded_premium_df = reinsurance_df.filter((col("data_date") >= start_date) & (col("data_date") <= end_date)) \
        .groupBy("org_id") \
        .agg(sum("ceded_premium").alias("total_ceded_premium"))

    # 计算自留保费
    retained_premium_df = total_premium_df.join(ceded_premium_df, on="org_id", how="inner") \
        .withColumn("retained_premium", col("total_premium") - col("total_ceded_premium"))

    # 计算自留保费占净资产比
    ratio_df = equity_df.join(retained_premium_df, on="org_id", how="inner") \
        .withColumn("avg_owner_equity", (col("start_owner_equity") + col("end_owner_equity")) / 2) \
        .withColumn("retained_premium_ratio", when(col("avg_owner_equity") != 0,
                                                   (col("retained_premium") / col("avg_owner_equity")) * 100)
                    .otherwise(0)) \
        .select("org_id", "retained_premium_ratio",
                F.lit(start_date).alias("start_date"),
                F.lit(end_date).alias("end_date"),
                F.lit(period_type).alias("period_type"))

    return ratio_df


# 周期计算调度函数
def calculate_all_indicators(start_date, end_date, period_type):
    """计算所有指标并合并结果"""
    # 计算各指标
    roe_df = calculate_roe(start_date, end_date, period_type)
    roa_df = calculate_roa(start_date, end_date, period_type)
    asset_turnover_df = calculate_asset_turnover(start_date, end_date, period_type)
    receivable_rate_df = calculate_receivable_premium_rate(start_date, end_date, period_type)
    receivable_turnover_df = calculate_receivable_turnover(start_date, end_date, period_type)
    embedded_value_df = calculate_embedded_value(start_date, end_date, period_type)
    assets_growth_df = calculate_assets_growth_rate(start_date, end_date, period_type)
    equity_growth_df = calculate_equity_growth_rate(start_date, end_date, period_type)
    underwriting_rate_df = calculate_underwriting_profit_rate(start_date, end_date, period_type)
    retained_ratio_df = calculate_retained_premium_ratio(start_date, end_date, period_type)

    # 合并所有指标结果
    result_df = roe_df \
        .join(roa_df, on=["org_id", "start_date", "end_date", "period_type"], how="inner") \
        .join(asset_turnover_df, on=["org_id", "start_date", "end_date", "period_type"], how="inner") \
        .join(receivable_rate_df, on=["org_id", "start_date", "end_date", "period_type"], how="inner") \
        .join(receivable_turnover_df, on=["org_id", "start_date", "end_date", "period_type"], how="inner") \
        .join(embedded_value_df, on=["org_id", "start_date", "end_date", "period_type"], how="inner") \
        .join(assets_growth_df, on=["org_id", "start_date", "end_date", "period_type"], how="inner") \
        .join(equity_growth_df, on=["org_id", "start_date", "end_date", "period_type"], how="inner") \
        .join(underwriting_rate_df, on=["org_id", "start_date", "end_date", "period_type"], how="inner") \
        .join(retained_ratio_df, on=["org_id", "start_date", "end_date", "period_type"], how="inner")

    # 关联机构名称
    result_with_name_df = result_df.join(
        org_df.select("org_id", "org_name").dropDuplicates(),
        on="org_id",
        how="left"
    )

    # 写入Paimon表，用于实时分析
    result_with_name_df.write \
        .format("paimon") \
        .mode("append") \
        .save("insurance_db.profitability_indicators")

    # 同时写入Doris，用于报表展示
    result_with_name_df.write \
        .format("jdbc") \
        .option("url", "jdbc:mysql://doris-host:9030/insurance_analysis") \
        .option("dbtable", "profitability_indicators") \
        .option("user", "username") \
        .option("password", "password") \
        .mode("append") \
        .save()

    return result_with_name_df


# 按周期调度计算
def schedule_calculation():
    """按日、周、月调度计算指标"""
    today = datetime.now().strftime("%Y-%m-%d")

    # 计算近7日指标
    seven_days_ago = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
    calculate_all_indicators(seven_days_ago, today, "7days")

    # 计算本月指标
    first_day_of_month = datetime.now().replace(day=1).strftime("%Y-%m-%d")
    calculate_all_indicators(first_day_of_month, today, "month")

    # 计算本年指标
    first_day_of_year = datetime.now().replace(month=1, day=1).strftime("%Y-%m-%d")
    calculate_all_indicators(first_day_of_year, today, "year")


# 执行调度
if __name__ == "__main__":

    # 1- 创建 SparkSession对象: 支持与HIVE的集成
    spark = SparkSession \
        .builder \
        .master("local[*]") \
        .appName("insurance_main") \
        .config("spark.sql.shuffle.partitions", 4) \
        .config("spark.sql.warehouse.dir", "hdfs://node1:8020/user/hive/warehouse") \
        .config("hive.metastore.uris", "thrift://node1:9083") \
        .enableHiveSupport() \
        .getOrCreate()

    schedule_calculation()
    spark.stop()